package mspbots.xero.client.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Builder;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.xero.client.core.data.BaseSyncService;
import mspbots.xero.client.oauth.AuthorizationCode;
import mspbots.xero.client.oauth.TokenState;
import mspbots.xero.common.RabbitKeys;
import mspbots.xero.common.RedisKeys;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Map;
import java.util.Optional;

/**
 * ScheduledAsyncService
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2019/6/19
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class GlobalScheduledTimer {

    private final Map<String, BaseSyncService> syncServiceMap;

    private final ReactiveStringRedisTemplate stringRedisTemplate;
    private final ReactiveRedisTemplate<String, Object> reactiveRedisTemplate;
    private final RabbitMessagingTemplate rabbitMessagingTemplate;
    private final ObjectMapper objectMapper;

    /**
     * 测试替换你开发的服务即可
     * (v instanceof ProjectsServiceImpl)
     * 修改请勿提交此文件
     */
    @Scheduled(fixedDelay = 1000 * 300)
    public void autoSyncTest() {
        this.reactiveRedisTemplate.scan(ScanOptions.scanOptions()
                .match(RedisKeys.OAUTH_REDIS_KEY_PREFIX + "*")
                .count(100).build())
                .flatMap(key -> this.reactiveRedisTemplate.opsForValue().get(key))
                .map(data -> this.objectMapper.convertValue(data, AuthorizationCode.class))
                .map(data -> TokenState.withDefault(data.getTenantId()).xeroTenants(data.getXeroTenants())
                        .setAccessToken(data.getAccessToken()))
                .filterWhen(tokenState -> this.stringRedisTemplate.opsForValue()
                        .size(RedisKeys.ON_OFF_TENANT_TASK_REDIS_KEY + tokenState.getTenantId())
                        .defaultIfEmpty(0L).map(r -> r == 0))
                .flatMap(tokenState -> Flux.fromStream(this.syncServiceMap.values().parallelStream())
                        .limitRate(8, 4)
                        .delayElements(Duration.ofSeconds(10))
                        .flatMap(v -> Flux.concatDelayError(
                                v.syncFast(tokenState),
                                v.syncSlower(tokenState),
                                v.syncSlow(tokenState)
                        )))
                .subscribe(res -> log.debug("Sync test success."),
                        err -> log.debug("Sync test error. msg: {}", err.getMessage()));
    }

    //@Scheduled(fixedDelay = 1000 * 30)
    public void autoSyncFast() {
        DistributedTask distributedTask = DistributedTask.builder().type("fast").build();
        this.distributedTaskLock(distributedTask)
                .doFinally(r -> this.stringRedisTemplate.opsForValue()
                        .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType(),
                                "false", Duration.ofSeconds(30)).subscribe())
                .block(Duration.ofSeconds(30));
    }

    //@Scheduled(cron = "0 0 0/2 * * ?")
    public void autoSyncSlower() {
        DistributedTask distributedTask = DistributedTask.builder().type("Slower").build();
        this.distributedTaskLock(distributedTask)
                .doFinally(r -> this.stringRedisTemplate.opsForValue()
                        .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType(),
                                "false", Duration.ofHours(2)).subscribe())
                .block(Duration.ofSeconds(10));
    }

    //@Scheduled(cron = "0 20 8 * * ?")
    public void autoSyncSlow() {
        DistributedTask distributedTask = DistributedTask.builder().type("slow").build();
        this.distributedTaskLock(distributedTask)
                .doFinally(r -> this.stringRedisTemplate.opsForValue()
                        .setIfAbsent(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType(),
                                "false", Duration.ofHours(12)).subscribe())
                .block(Duration.ofSeconds(10));
    }

    private Mono<Void> distributedTaskLock(DistributedTask distributedTask) {
        return this.stringRedisTemplate.opsForValue()
                .get(RedisKeys.TASK_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType())
                .map(Boolean::parseBoolean).defaultIfEmpty(true)
                .filter(r -> r)
                .flatMapMany(r -> this.onSendTask(distributedTask))
                .doOnNext(tokenState -> log.debug("Sync send distributed [{}] ,tenant [{}]",
                        distributedTask.getType(), tokenState.getTenantId()))
                .then();
    }

    private Flux<TokenState> onSendTask(DistributedTask distributedTask) {
        return this.reactiveRedisTemplate.scan(ScanOptions.scanOptions()
                .match(RedisKeys.OAUTH_REDIS_KEY_PREFIX + "*")
                .count(100).build())
                .flatMap(key -> this.reactiveRedisTemplate.opsForValue().get(key))
                .map(data -> this.objectMapper.convertValue(data, AuthorizationCode.class))
                .map(data -> TokenState.withDefault(data.getTenantId()).setAccessToken(data.getAccessToken()))
                .filterWhen(tokenState -> this.stringRedisTemplate.opsForValue()
                        .size(RedisKeys.ON_OFF_TENANT_TASK_REDIS_KEY + tokenState.getTenantId())
                        .defaultIfEmpty(0L).map(r -> r == 0))
                .doOnNext(tokenState -> this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                        RabbitKeys.DISTRIBUTED_TASK_QUEUE_KEY, distributedTask.setTokenState(tokenState)));
    }

    @RabbitListener(queues = "xero.distributed.task.sync", concurrency = "10")
    public void syncTaskRunning(DistributedTask distributedTask) {

        String lockKey = RedisKeys.TENANT_RUNNING_LOCK_ANCHOR_REDIS_KEY + distributedTask.getType()
                + ":" + distributedTask.getTokenState().getTenantId();
        Optional<Long> lockCountOpt = this.stringRedisTemplate.opsForValue().size(lockKey).blockOptional();

        if (lockCountOpt.isPresent() && lockCountOpt.get() > 0) {
            log.info("===Sync [{}] Task [{}] Running===",
                    distributedTask.getTokenState().getTenantId(), distributedTask.getType());
        } else {
            this.reactiveRedisTemplate.opsForValue()
                    .setIfAbsent(lockKey, distributedTask.getTokenState().getTenantId(), Duration.ofMinutes(60))
                    .flatMapMany(res -> Flux.fromStream(this.syncServiceMap.keySet().parallelStream()))
                    .delayElements(Duration.ofMillis(200))
                    .concatMapDelayError(key -> {
                        var service = this.syncServiceMap.get(key);
                        if ("fast".equals(distributedTask.getType())) {
                            return service.syncFast(distributedTask.getTokenState())
                                    .onErrorResume(err -> {
                                        log.error("Sync [{}] task [{}]  error, msg: [{}]",
                                                distributedTask.getTokenState().getTenantId(),
                                                key, err.getMessage());
                                        return Mono.empty();
                                    });
                        } else if ("slow".equals(distributedTask.getType())) {
                            return service.syncSlow(distributedTask.getTokenState())
                                    .onErrorResume(err -> {
                                        log.error("Sync [{}] task [{}]  error, msg: [{}]",
                                                distributedTask.getTokenState().getTenantId(),
                                                key, err.getMessage());
                                        return Mono.empty();
                                    });
                        } else {
                            return service.syncSlower(distributedTask.getTokenState())
                                    .onErrorResume(err -> {
                                        log.error("Sync [{}] task [{}]  error, msg: [{}]",
                                                distributedTask.getTokenState().getTenantId(),
                                                key, err.getMessage());
                                        return Mono.empty();
                                    });
                        }
                    })
                    .doFinally(v -> this.reactiveRedisTemplate.opsForValue().delete(lockKey).subscribe())
                    .subscribe(result -> log.debug("===Sync [{}] Task [{}] stop===",
                            distributedTask.getTokenState().getTenantId(), distributedTask.getType()),
                            err -> log.error("Sync [{}] {} task error, msg: [{}]",
                                    distributedTask.getTokenState().getTenantId(), distributedTask.getType(),
                                    err.getMessage()));
        }
    }

    @Data
    @Builder
    static class DistributedTask {
        private String type;
        private TokenState tokenState;

        public DistributedTask setTokenState(TokenState tokenState) {
            this.tokenState = tokenState;
            return this;
        }
    }
}
